package com.hadoop.project.gen.maodaligendata;

import com.hadoop.project.entity.hbase.HBaseUser;
import com.hadoop.project.entity.hbase.Tab;
import com.hadoop.project.entity.hbase.UserLog;
import com.hadoop.project.entity.hbase.Video;
import com.hadoop.project.gen.maodaligendata.HBaseUtils;
import com.hadoop.project.gen.maodaligendata.GenConstant;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author alexmaodali
 * @usage:生成hbase数据
 */
public class GenHBaseData {

    public static void execute(String tableName, Integer genVideoLength, Integer genUserLength, Integer genUserLogLength) throws Exception {
        Connection conn = HBaseUtils.getConnection();
        Table table     = conn.getTable(TableName.valueOf(tableName));

        List<String> tidList = new ArrayList<>();
        List<String> uidList = new ArrayList<>();
        List<String> vidList = new ArrayList<>();
        Map<String, String> vidMapFlip = new HashMap<>();

        // 生成Tab数据
        for (int i = 0; i < GenConstant.tabList.length; i++) {
            Tab tab = Tab.builder().tid("tid-" + (i + 1)).tname(GenConstant.tabList[i]).build();
            tidList.add(tab.getTid());
            // 插入数据到HBase
            Put put = new Put(Bytes.toBytes(tab.getTid()));
            put.addColumn(Bytes.toBytes("tab"), Bytes.toBytes("tid")    , Bytes.toBytes(tab.getTid()));
            put.addColumn(Bytes.toBytes("tab"), Bytes.toBytes("tabName"), Bytes.toBytes(tab.getTname()));
            table.put(put);
        }

        // 生成User数据
        for (int i = 0; i < genUserLength; i++) {
            HBaseUser user = HBaseUser.builder().uid(GenConstant.genUid(10)).nickName(GenConstant.genNickName()).sex(GenConstant.genSex()).age(GenConstant.genAge()).province(GenConstant.genProvince()).build();
            uidList.add(user.getUid());
            // 插入数据到HBase
            Put put = new Put(Bytes.toBytes(user.getUid()));
            put.addColumn(Bytes.toBytes("user"), Bytes.toBytes("uid")     , Bytes.toBytes(user.getUid()));
            put.addColumn(Bytes.toBytes("user"), Bytes.toBytes("nickname"), Bytes.toBytes(user.getNickName()));
            put.addColumn(Bytes.toBytes("user"), Bytes.toBytes("sex")     , Bytes.toBytes(user.getSex()));
            put.addColumn(Bytes.toBytes("user"), Bytes.toBytes("age")     , Bytes.toBytes(user.getAge()));
            put.addColumn(Bytes.toBytes("user"), Bytes.toBytes("province"), Bytes.toBytes(user.getProvince()));
            table.put(put);
        }

        // 生成Video数据
        for (int i = 0; i < genVideoLength; i++) {
            Video video = Video.builder().vid(GenConstant.genLongTypeVID(10)).tabs(GenConstant.genTabs(tidList)).author(GenConstant.genAuthor(uidList)).flip(GenConstant.genFlip()).build();
            vidList.add(video.getVid());
            vidMapFlip.put(video.getVid(), video.getFlip());
            // 插入数据到HBase
            Put put = new Put(Bytes.toBytes(video.getVid()));
            put.addColumn(Bytes.toBytes("video"), Bytes.toBytes("vid")   , Bytes.toBytes(video.getVid()));
            put.addColumn(Bytes.toBytes("video"), Bytes.toBytes("tabs")  , Bytes.toBytes(video.getTabs()));
            put.addColumn(Bytes.toBytes("video"), Bytes.toBytes("author"), Bytes.toBytes(video.getAuthor()));
            put.addColumn(Bytes.toBytes("video"), Bytes.toBytes("flip")  , Bytes.toBytes(video.getFlip()));
            table.put(put);
        }

        // 生成UserLog数据
        for (int i = 0; i < genUserLogLength; i++) {
            UserLog userLog = UserLog.builder().uid(GenConstant.genAuthor(uidList)).id("userLog-" + (i + 1)).vid(GenConstant.genVidInUserLog(vidList)).videoTime(GenConstant.genVideoTime(10)).build();
            String flip = vidMapFlip.get(userLog.getVid());
            userLog.setVideoTime(flip);
            // 插入数据到HBase
            Put put = new Put(Bytes.toBytes("userlog-" + userLog.getId()));
            put.addColumn(Bytes.toBytes("userlog"), Bytes.toBytes("id")       , Bytes.toBytes(userLog.getId()));
            put.addColumn(Bytes.toBytes("userlog"), Bytes.toBytes("uid")      , Bytes.toBytes(userLog.getUid()));
            put.addColumn(Bytes.toBytes("userlog"), Bytes.toBytes("vid")      , Bytes.toBytes(userLog.getVid()));
            put.addColumn(Bytes.toBytes("userlog"), Bytes.toBytes("videoTime"), Bytes.toBytes(userLog.getVideoTime()));
            table.put(put);
        }
        // 关闭流
        table.close();
        conn.close();
    }
}
